package org.company.mars.zookeeper.monitor;

import java.util.Arrays;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.data.Stat;

public class DataMonitor implements StatCallback,Watcher {
    
    private ZooKeeper zk;
    
    private String znode;
    
    public boolean dead;
    
    private DataMonitorListener listener;
    
    private byte[] prevData;
    
    private Watcher chainedWatcher;
    
    public interface DataMonitorListener {
        
        /**
         * The existence status of the node has changed
         * @param data
         */
        void reboot(byte[] data);
        
        /**
         * 
         * @param rc 
         *          The Zookeeper reason code
         */
        void closing(int rc);
    }
    
    public DataMonitor(ZooKeeper zk,
            String znode,
            DataMonitorListener listener) {
        this.dead = false;
        this.zk = zk;
        this.znode = znode;
        this.listener = listener;
        this.zk.exists(this.znode, true, this, null);
    }
    
    @SuppressWarnings("deprecation")
    @Override
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        // TODO Auto-generated method stub
        boolean exists;
        switch(rc) {
        case Code.Ok:
            exists = true;
            break;
        case Code.NoNode:
            exists = false;
            break;
        case Code.SessionExpired:
        case Code.NoAuth:
            this.dead = true;
            listener.closing(rc);
            return;
        default:
            zk.exists(this.znode, true, this, null);
            return;
        }
        
        byte[] data = null;
        if(exists) {
            try {
                data = zk.getData(this.znode, false, null);
            } catch (KeeperException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        
        if((data == null && this.prevData != data)
                || (data != null && !Arrays.equals(data, this.prevData))) {
            if(Arrays.equals(data, "quit".getBytes())) {
                listener.closing(KeeperException.Code.Ok);
                return;
            } else {
                listener.reboot(data);
            }
            this.prevData = data;
        }
    }
    
    @SuppressWarnings({ "deprecation", "incomplete-switch" })
    @Override
    public void process(WatchedEvent event) {
        // TODO Auto-generated method stub
        String path = event.getPath();
        if(event.getType() == EventType.None) {
            switch(event.getState()) {
            case SyncConnected:
                break;
            case Expired:
                this.dead = true;
                this.listener.closing(KeeperException.Code.SessionExpired);
                break;
            }
        } else {
            if(path != null && path.equals(this.znode)) {
                zk.exists(path, true, this, null);
            }
        }
        
        if(this.chainedWatcher != null) {
            this.chainedWatcher.process(event);
        }
    }
}
